#include "handy/handy/handy.h"
#include <algorithm>
#include <stdio.h>
#include <string>
#include <stdint.h>
#include <arpa/inet.h>
#include <map>
#include <sstream>
#include <vector>
#include "xudp.pb.h"
#include "common.h"
using namespace std;
using namespace handy;
int g_una = 0;
int g_max = 0;

Ip4Addr g_client;
struct PackageXInfo{
    int64_t x_time;
    int64_t x_num;
};

std::map<int, PackageXInfo> g_x_info;
std::map<int, int> g_out_of_order;

void check_una(){
    for( auto id = g_una; id <= g_max; id++){
        if(g_out_of_order.find(id) != g_out_of_order.end()){
            g_una = id+1;
        } else {
            break;
        }
    }
}

void handle_package(xpackage& p) {
	if (p.cmd() == CMD_DATA) {
		//info("size:%d\n", p.msg_size());
		vector<int> msg_id;
		for (int i = 0; i < p.msg_size(); i++) {
			const xmsg m = p.msg(i);
			msg_id.push_back(m.id());
		}
		std::sort(msg_id.begin(), msg_id.end());
        stringstream ss;
        for (auto id :msg_id){
            ss << id << "_";
        }
        int original_una = g_una;

		for (auto id : msg_id) {
            g_x_info.erase(id);
            if(g_max < id){
                g_max = id;
            }

			if (id > g_una){
                g_out_of_order[id] = 1;
			}

			if (id == g_una) {
                g_out_of_order.erase(id);
				info("id:%d,time:%ld", id,util::timeMilli());
				g_una++;
			}
		}
        check_una();
        info("max:%d, original:%d, una:%d, msg:%s", g_max, original_una, g_una, ss.str().c_str());
	}
}
int main() {
	setloglevel("INFO");
    setlogfile("test_server_partial_udp");
	EventBase base;
	Signal::signal(SIGINT, [&] { base.exit(); });

	UdpServerPtr s = handy::UdpServer::startServer(&base, "", SERVER_PORT);
	exitif(s == NULL, "start udp server failed");

	s->onMsg([&](const UdpServerPtr& p, Buffer buf, Ip4Addr peer) {
		xpackage package;
		bool suc = package.ParseFromArray(buf.data(), buf.size());
		if (suc) {
			handle_package(package);
		}

        g_client = peer;
        if(g_una==MAX_MSG_ID+1){
            xpackage p;
            p.set_cmd(CMD_FINISH);
            char b[100];
            bool suc = p.SerializeToArray(b, sizeof(b));
            if(!suc){
                fatal("Serialize failed");
            }
            s->sendTo(b, p.ByteSize(), g_client);
            info("exited");
            base.exit();    
        }
	});

    base.runAfter(0, [&](){
        check_una();
        if(g_una==MAX_MSG_ID+1){
            xpackage p;
            p.set_cmd(CMD_FINISH);
            char b[100];
            bool suc = p.SerializeToArray(b, sizeof(b));
            if(!suc){
                fatal("Serialize failed");
            }
            s->sendTo(b, p.ByteSize(), g_client);
            info("exited");
            base.exit();    
        }
        xpackage p;
        p.set_cmd(CMD_GET_DATA);
        stringstream ss;
        for(int i = g_una; i < g_max; i++){
            if(g_out_of_order.find(i) != g_out_of_order.end()){
                continue;
            }
            int lost_id = i;
            if(g_x_info[lost_id].x_time == 0 || g_x_info[lost_id].x_time > util::timeMilli()) {
                g_x_info[lost_id].x_num++;
                g_x_info[lost_id].x_time = util::timeMilli() + (g_x_info[lost_id].x_num)*2*MSG_INTERVAL;
                p.add_lost_id(lost_id);
                ss <<lost_id <<"_";
            }
        }

        if(p.lost_id_size() == 0){
            return;
        } 
        info("lost package:%s",ss.str().c_str());

        char b[510]={};
        bool suc = p.SerializeToArray(b, sizeof(b));
        if(!suc){
            fatal("Serialize failed");
        }
        s->sendTo(b, p.ByteSize(), g_client);
    }, 20);
	base.loop();
	info("program exited");
}
